import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Scanner;
import java.util.Random;
import java.util.StringTokenizer;

/**
 * The Paxos Peer that assumes multiple roles
 */

public class Paxos1Peer {

	int pid; /* Process PID */
//	PaxosMessage highestAcceptedProposal; /* Last proposal it accepted */
//	PaxosMessage highestPrepareResponded; /* Most recent proposal it heard */
	int port; /* UDP port to run on */
	String remoteHost;
	String logFile;
	float dropProbability;
	DatagramSocket pSocket;
	ArrayList<Integer > upSet;
	int currentRole;
	int currentLeader;
	
	int currentInstanceId;
	// leader DS
	ArrayList<Integer> instancePrepareProposalNum = new ArrayList<Integer>();
	ArrayList<String> instanceProposalVal = new ArrayList<String>();

	//acceptor DS
	ArrayList<Integer> instanceHighestAcceptProposalNum = new ArrayList<Integer>();
	ArrayList<String> instanceHighestAcceptProposalVal = new ArrayList<String>();
	
	ArrayList<Integer> instanceHighestPrepareProposalNum = new ArrayList<Integer>();

	ArrayList<Integer> clientAccounts = new ArrayList<Integer>();
	
	ArrayList<InetSocketAddress> clientSocket = new ArrayList<InetSocketAddress>();
	byte[] buf = new byte[PaxosConsts.MESG_SIZE];
	
	public PaxosPeer() {
		// TODO Auto-generated constructor stub
		System.out.println("DEBUG"+"inside def constructor");
		logFile = pid+".log";
		dropProbability = 0;
		currentInstanceId=0;
		
	}
	
	public PaxosPeer(int pid, int port, float dropProbability, int currentRole, String remoteHost, ArrayList<Integer> upSet)
	{
		this();
		this.pid = pid;
		this.port = port;
		this.dropProbability = dropProbability;
		this.upSet = upSet;
		this.currentRole = currentRole;
		this.remoteHost = remoteHost;
		System.out.println("DEBUG: all process information");
		try
		{
			pSocket = new DatagramSocket(port);
			System.out.println("DEBUG what port"+port);
		}
		catch(SocketException se)
		{
			System.err.println(se.getMessage());
		}
		System.out.println("pid:"+this.pid+" port:"+this.port+" dp:"+this.dropProbability+" currentrole:"+this.currentRole+" host:"+this.remoteHost+" upset:"+this.upSet.toString());
		
		for (int i=0;i<PaxosConsts.ARRAYLIST_SIZE;i++)
		{
			instancePrepareProposalNum.add(i, PaxosConsts.NO_ELEMENT);
			instanceProposalVal.add(i, PaxosConsts.NO_ELE_STR);
			instanceHighestAcceptProposalNum.add(i, PaxosConsts.NO_ELEMENT);
			instanceHighestAcceptProposalVal.add(i, PaxosConsts.NO_ELE_STR);
			instanceHighestPrepareProposalNum.add(i, PaxosConsts.NO_ELEMENT);
			clientSocket.add(i, null);
			clientAccounts.add(i, PaxosConsts.NO_ELEMENT);
		}
	}
	
	public boolean dropPacket()
	{
		// Drop packet based on peer probability
		// Say drop probability is .3
		// Generate a random number between [0,9). If the random number is less than 3 return true.
		// So indirectly we generate a number with probability of 33%. 
		Random rand = new Random();
		if (rand.nextInt(PaxosConsts.PROBABILTY_BOUND) < (dropProbability*PaxosConsts.PROBABILTY_BOUND))
			return true;
		else
			return false;
	}
	
	
	/*
	 * Roles that can be played
	 */

	public int playAcceptor()
	{
		int protocolState = PaxosConsts.CONTINUE;
		
		try
		{
			pSocket.setSoTimeout(PaxosConsts.PEER_TIMEOUT_BOUND);
		}
		catch (SocketException e) {
			// TODO: handle exception
		}
		do
		{
			
			DatagramPacket inboundMesg = new DatagramPacket(buf,buf.length);
			try
			{
				Arrays.fill(inboundMesg.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
				pSocket.receive(inboundMesg);
				// Drop messages based on the drop probability
				if(dropPacket())
				{
					System.err.println("PID: "+pid+" - Dropping packet recieved from");
				}
				else
				{
					
					PaxosMessage inboundPaxosMesg = new PaxosMessage(inboundMesg.getData());
					System.out.println("PID: "+pid+" - Received packet from: "+inboundPaxosMesg.senderId);
					if (inboundPaxosMesg.msgType.equals(PaxosConsts.PREPARE)) 
					{
						System.out.println("DEBUG got mesg type prepare..processprepare");
						processPrepare(inboundPaxosMesg);

					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.ACCEPT))
					{
						System.out.println("DEBUG got mesg type ACCEPT..processaccept");
						processAccept(inboundPaxosMesg);
					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.LEARN))
					{
						System.out.println("DEBUG got mesg type learn..processlearn");
						processLearn(inboundPaxosMesg.proposalValue,inboundPaxosMesg.clientId);
					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.CLIENT_REQ))
					{
						System.out.println("PID: "+pid+" - Received CLIENT_REQ. ignoring...");
					}
					else if (inboundPaxosMesg.msgType.equals(PaxosConsts.EXIT))
					{
						System.out.println("PID: "+pid+" - Received Exit. session done");
						protocolState = PaxosConsts.SESSION_DONE;
					}
					
				}
			}
			catch (SocketTimeoutException e) 
			{
				// TODO: handle exception
				protocolState = PaxosConsts.LEADER_DIED;
			}
			catch (IOException e) 
			{
				// TODO: handle exception
			}
		}while(protocolState == PaxosConsts.CONTINUE);
		
		
		return protocolState;
	}
	
	private void processLearn(String proposalValGot, int clientID) {
		// TODO Auto-generated method stub
		// WRITE TO DISK CLIENT RESPONSE
		if (proposalValGot.indexOf(PaxosConsts.BALANCE) != -1)
		{
			
		}
		if (proposalValGot.indexOf(PaxosConsts.WITHDRAW) != -1)
		{
			StringTokenizer withdrawTokens = new StringTokenizer(proposalValGot);
			withdrawTokens.nextToken(); 
			clientAccounts.set(clientID,clientAccounts.get(clientID)-Integer.parseInt(withdrawTokens.nextToken()));
			
		}
		if (proposalValGot.indexOf(PaxosConsts.DEPOSIT) != -1)
		{
			StringTokenizer depositTokens = new StringTokenizer(proposalValGot);
			depositTokens.nextToken(); 
			clientAccounts.set(clientID,clientAccounts.get(clientID)+Integer.parseInt(depositTokens.nextToken()));
			
		}
		
	}

	private void processPrepare(PaxosMessage inboundPrepareMsg) 
	{
		
		int proposalNumGot = inboundPrepareMsg.proposalNum;
		int instanceIDGot = inboundPrepareMsg.instanceId;
		int clientID = inboundPrepareMsg.clientId;
		byte promise[] = null;
		System.out.println("DEBUG in process prepare for :"+clientID);
		if (proposalNumGot <= instanceHighestPrepareProposalNum.get(instanceIDGot))
		{
			// TODO nothing
			System.out.println("DEBUG got propnum <= highest etc..");
			promise = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PREPARE_REJECT,pid,instanceIDGot,instanceHighestPrepareProposalNum.get(instanceIDGot),
					PaxosConsts.NO_ELE_STR,
					null,instanceHighestAcceptProposalNum.get(instanceIDGot),instanceHighestAcceptProposalVal.get(instanceIDGot),PaxosConsts.NO_ELE_STR,
					clientID);

		}
		else
		{
			System.out.println("DEBUG Got a higher prepropnum: "+proposalNumGot+" for instanceid:"+instanceIDGot);
			System.out.println("DEBUG compared with :"+instanceHighestPrepareProposalNum.get(instanceIDGot));
			instanceHighestPrepareProposalNum.set(instanceIDGot,proposalNumGot);
			promise = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PROMISE,pid,instanceIDGot,proposalNumGot,PaxosConsts.NO_ELE_STR,
					null,instanceHighestAcceptProposalNum.get(instanceIDGot),instanceHighestAcceptProposalVal.get(instanceIDGot),PaxosConsts.NO_ELE_STR,
					clientID);
		}
		DatagramPacket promisePacket = null; 
			
		try
		{
			System.out.println("DEBUG cmdlist sending: "+(new String(promise)));
			promisePacket = new DatagramPacket(promise, promise.length, InetAddress.getByName(remoteHost),upSet.get(currentLeader));
			pSocket.send(promisePacket);
			System.out.println("DEBUG send promise");
		}
		catch (UnknownHostException he) 
		{
			System.err.println(he.getMessage());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	private void processAccept(PaxosMessage inboundAcceptMsg) 
	{
		int proposalNumGot = inboundAcceptMsg.proposalNum;
		int instanceIDGot = inboundAcceptMsg.instanceId;
		String proposalValGot = inboundAcceptMsg.proposalValue;
		int clientID = inboundAcceptMsg.clientId;
		if (proposalNumGot < instanceHighestPrepareProposalNum.get(instanceIDGot))
		{
			// TODO nothing
		}
		else
		{
			// update current instance highs
			instanceHighestAcceptProposalNum.set(instanceIDGot,proposalNumGot);
			instanceHighestAcceptProposalVal.set(instanceIDGot,proposalValGot);
			String respToClient = tentativeClientAccount(proposalValGot,clientID);
			byte chosen[] = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CHOSEN,pid,instanceIDGot,proposalNumGot,proposalValGot,
					null,PaxosConsts.NO_ELE_INT,PaxosConsts.NO_ELE_STR,respToClient,clientID);
			
			DatagramPacket chosenPacket = null; 
			
			
			try
			{
				chosenPacket = new DatagramPacket(chosen, chosen.length, InetAddress.getByName(remoteHost),upSet.get(currentLeader));
				pSocket.send(chosenPacket);
				System.out.println("DEBUG send chosen");
			}
			catch (UnknownHostException he) 
			{
				System.err.println(he.getMessage());
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	private String tentativeClientAccount(String proposalValGot, int clientID) {
		// TODO Auto-generated method stub
		
		if (proposalValGot.indexOf(PaxosConsts.BALANCE) != -1)
		{
			return PaxosConsts.BALANCE_RESPONSE+clientAccounts.get(clientID)+":"+clientAccounts.get(clientID);
		}
		if (proposalValGot.indexOf(PaxosConsts.WITHDRAW) != -1)
		{
			StringTokenizer withdrawTokens = new StringTokenizer(proposalValGot);
			withdrawTokens.nextToken(); 
			int tentativeBal = clientAccounts.get(clientID)-Integer.parseInt(withdrawTokens.nextToken());
			return PaxosConsts.WITHDRAW_RESPONSE+":"+tentativeBal;
		}
		if (proposalValGot.indexOf(PaxosConsts.DEPOSIT) != -1)
		{
			StringTokenizer depositTokens = new StringTokenizer(proposalValGot);
			depositTokens.nextToken(); 
			int tentativeBal = clientAccounts.get(clientID)+Integer.parseInt(depositTokens.nextToken());
			return PaxosConsts.DEPOSIT_RESPONSE+":"+tentativeBal;
		}
		
		
		return null;
	}

	public void processChosen(DatagramPacket chosenPacket)
	{
		PaxosMessage chosenMesg = (new PaxosMessage(chosenPacket.getData()));
		
		byte [] learnMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.LEARN, pid, 
				currentInstanceId, chosenMesg.proposalNum, chosenMesg.proposalValue, null, chosenMesg.highAcceptProposalNum, 
				chosenMesg.highAcceptProposalValue,chosenMesg.respToClient,chosenMesg.clientId);
		
		for (int j=0;j<upSet.size();j++)
		{
 			try
			{
 				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
 				{
					pSocket.send(new DatagramPacket(learnMsg,learnMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
					System.out.println("sent learn...to "+j);
 				}
			}
			catch(IOException io)
			{
				System.err.println(io.getMessage());
			}
		}
		
		String respToClient = chosenMesg.respToClient;
		
		
		StringTokenizer respAndValue = new StringTokenizer(respToClient,":");
		String resp = respAndValue.nextToken();
		String value = respAndValue.nextToken();
		String finalClientResp = "";
		if (resp.equals(PaxosConsts.WITHDRAW_RESPONSE) || resp.equals(PaxosConsts.DEPOSIT_RESPONSE))
		{
			finalClientResp = resp;
		}
		else
		{
			finalClientResp = resp+value;	
		}
		byte [] respClientMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CLIENT_RSP, pid, 
				currentInstanceId, chosenMesg.proposalNum, chosenMesg.proposalValue, null, chosenMesg.highAcceptProposalNum, 
				chosenMesg.highAcceptProposalValue,finalClientResp,chosenMesg.clientId);
		try
		{
			InetSocketAddress returnAddress = clientSocket.get(chosenMesg.clientId);
			pSocket.send(new DatagramPacket(respClientMsg,respClientMsg.length,returnAddress));
			System.out.println("sent final response...to client");
		}
		catch(IOException io)
		{
			System.err.println(io.getMessage());
		}
	}
	
	public int processPromise(ArrayList<String> allInboundPromises,int numPromiseGot)
	{
		int numChosenReceived = 0;
	    int numChosenExpected = 0;
	    int numUpPeers = 0;
	    int highestPrepareProposalNum=PaxosConsts.NO_ELEMENT;
	    int highestAcceptProposalNum=PaxosConsts.NO_ELE_INT;
	    String highestAcceptProposalVal=PaxosConsts.NO_ELE_STR;
	    int highestProposalPacket=PaxosConsts.NO_ELEMENT;
	    String msgTypeGot = "";
	    int newProposalNum = PaxosConsts.NO_ELEMENT;
	    int proposalNum = PaxosConsts.NO_ELEMENT;
	    String newProposalVal = PaxosConsts.NO_ELE_STR;
	    int newInstanceId = PaxosConsts.NO_ELEMENT;
	    int clientID = PaxosConsts.NO_ELEMENT;
	    System.out.println("DEBUG in process promise");
	    System.out.println("DEBUG size of allinoubd: "+allInboundPromises.size());
	    for (int i=0;i<numPromiseGot;i++)
	    {
	    	System.out.println("DEBUG promise mesg: "+i+" --- "+new String(allInboundPromises.get(i)));
	    	PaxosMessage promiseMsg= new PaxosMessage(allInboundPromises.get(i));
	    	clientID = promiseMsg.clientId;
	    	if (promiseMsg.highAcceptProposalNum > highestAcceptProposalNum)
	    	{
	    		highestProposalPacket=i;
	    		highestAcceptProposalNum = promiseMsg.highAcceptProposalNum;
	    		highestAcceptProposalVal =promiseMsg.highAcceptProposalValue;
	    		msgTypeGot = promiseMsg.msgType;
	    		proposalNum = promiseMsg.proposalNum;
	    		newInstanceId = promiseMsg.instanceId;
	    		if (msgTypeGot.equals(PaxosConsts.PREPARE_REJECT))
	    		{
	    			highestPrepareProposalNum = promiseMsg.proposalNum;
	    		}
	    	}
	    }
	    System.out.println(":"+msgTypeGot+":");
	    if (msgTypeGot.equals(PaxosConsts.PROMISE))
	    {
	    	System.out.println("DEBUG got PROMISE not REJECT");
	    	newProposalNum = proposalNum;
		    if (highestAcceptProposalNum == PaxosConsts.NO_ELEMENT)
		    {
		    	newInstanceId = new PaxosMessage(allInboundPromises.get(highestProposalPacket)).instanceId;
		    	newProposalVal = instanceProposalVal.get(newInstanceId);
		    }
		    else
		    {
		    	newProposalVal =highestAcceptProposalVal;
		    }
	    }
	    else if (msgTypeGot.equals(PaxosConsts.PREPARE_REJECT))
	    {
	    	// rejected prepare
	    	System.out.println("DEBUG got REJECT");
	    	instancePrepareProposalNum.set(newInstanceId,highestPrepareProposalNum);
	    	return PaxosConsts.PREPARE_REJECTED;
	    }
		byte [] acceptMsg = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.ACCEPT, pid, newInstanceId, newProposalNum, newProposalVal, null, 
				PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,clientID);
		for (int j=0;j<upSet.size();j++)
		{
			if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
			{
				try
				{
					pSocket.send(new DatagramPacket(acceptMsg,acceptMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
					numChosenExpected++;
					numUpPeers++;
				}
				catch(IOException io)
				{
					System.err.println(io.getMessage());
				}
			}
		}
		DatagramPacket inboundChosenPacket = new DatagramPacket(buf,buf.length);
	    while (numChosenReceived < numChosenExpected)
	    {
	    	try 
	    	{
	    		Arrays.fill(inboundChosenPacket.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
				pSocket.receive(inboundChosenPacket);
		    	if(!dropPacket())
		    	{
	    			
		    		numChosenReceived++;
		    	}
		    	else
		    	{
		    		System.err.println("PID: "+pid+" - Dropping packet recieved from");
		    	}

			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	    	
	    }
	    processChosen(inboundChosenPacket);
	    return PaxosConsts.CONTINUE;

	}
	
	public int processClientReq(PaxosMessage inboundClientMesg)
	{
		System.out.println("DEBUG inside creq");
	    DatagramPacket inboundPaxosMessage = new DatagramPacket(buf,buf.length);
	    ArrayList<String> allInboundPromises = new ArrayList<String>();
	    String cmd = "";
	    int numPromiseReceived = 0;
	    int numPromiseExpected = 0;
	    int numCmds = inboundClientMesg.cmdParamList.size();
	    int numUpPeers = 0;
	    int protocolState = PaxosConsts.CONTINUE;
	    int i=0;
	    int promiseRetVal = PaxosConsts.CONTINUE;
	    System.out.println("DEBUG inboundmesg: "+inboundClientMesg.cmdParamList.toString()+" numcmd: "+numCmds);
	    System.out.println("DEBUG cid:"+inboundClientMesg.senderId);
//	    for (int j=0;j<PaxosConsts.ARRAYLIST_SIZE;j++)
//		{
//	    	allInboundPromises.add(null);
//		}
	    while (i< numCmds) 
	    {
			cmd = inboundClientMesg.cmdParamList.get(i);
		    System.out.println("DEBUG cid:"+inboundClientMesg.senderId);
			int clientID = inboundClientMesg.senderId;
			if (cmd.equals(PaxosConsts.EXIT))
			{
				byte[] exitMsg = PaxosConsts.EXIT.getBytes();
	 			for (int j=0;j<upSet.size();j++)
	 			{
		 			try
					{
		 				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
		 				{
							pSocket.send(new DatagramPacket(exitMsg,exitMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
							System.out.println("sent exit...to "+j);
		 				}
					}
					catch(IOException io)
					{
						System.err.println(io.getMessage());
					}
	 			}
	 			return PaxosConsts.SESSION_DONE;

			}
			else
			{
				if (currentInstanceId == 0 || promiseRetVal != PaxosConsts.PREPARE_REJECTED)
				{
					currentInstanceId++;
				}
				int currPreparePropNum = instancePrepareProposalNum.get(currentInstanceId);
				instancePrepareProposalNum.set(currentInstanceId, ++currPreparePropNum);
				instanceProposalVal.set(currentInstanceId, cmd);
				
				byte [] prepareMsg  = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.PREPARE, pid, 
						currentInstanceId, currPreparePropNum, PaxosConsts.NO_ELE_STR, null, PaxosConsts.NO_ELE_INT, 
						PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,clientID);
				System.out.println("DEBUG clientID before upset: "+clientID);
				System.out.println("DEBUG not an exit.. and now upsetting.."+upSet.toString());
				for (int k=0;k<upSet.size();k++)
				{
					System.out.println("DEBUG "+upSet.get(k));
				}
				numUpPeers = 0;
				numPromiseExpected=0;
	 			for (int j=0;j<upSet.size();j++)
	 			{
		 			try
					{
		 				if (upSet.get(j) != PaxosConsts.DEAD_PROCESS_PORT && j!=pid)
		 				{
							pSocket.send(new DatagramPacket(prepareMsg,prepareMsg.length,InetAddress.getByName(remoteHost),upSet.get(j)));
							numPromiseExpected++;
							numUpPeers++;
							System.out.println("sent prepare...to "+j);
		 				}
					}
					catch(IOException io)
					{
						System.err.println(io.getMessage());
					}
	 			}
	 			System.out.println("DEBUG promised expected: "+numPromiseExpected+" numuppeers: "+numUpPeers);
	 			while (numPromiseReceived < numPromiseExpected)
	 			{
	 				try 
	 				{
	 					Arrays.fill(inboundPaxosMessage.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
	 					System.out.println("DEBUG promise mesg: restted --- "+new String(inboundPaxosMessage.getData()));
						pSocket.receive(inboundPaxosMessage);
						System.out.println("DEBUG promise mesg: received --- "+new String(inboundPaxosMessage.getData()));
						allInboundPromises.add(new String(inboundPaxosMessage.getData()));
	 		 			System.out.println("DEBUG promise mesg: reprint jas1 --- "+new String(allInboundPromises.get(0)));
					} catch (IOException e) 
					{
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
	 				if(!dropPacket())
	 		    	{
	 		    		++numPromiseReceived;
	 		    	}
	 		    	else
	 		    	{
	 		    		System.err.println("PID: "+pid+" - Dropping packet recieved from");
	 		    	}
	 			}
	 			promiseRetVal = processPromise(allInboundPromises,numPromiseReceived);
	 			if (promiseRetVal != PaxosConsts.PREPARE_REJECTED)
	 			{
	 				numPromiseReceived=0;
	 				i++;
	 			}
			}
	    }
	    return protocolState;
	}
	
	public int playLeader()
	{
		int protocolState = PaxosConsts.CONTINUE;		// TODO: should i be dropping client requests
		DatagramPacket inboundMesg = new DatagramPacket(buf,buf.length);
		
		do
		{
			try
			{
				Arrays.fill(inboundMesg.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
				System.out.println("DEBUG waiting in leader: "+pSocket.getLocalPort()+" "+pSocket.getPort()+" "+pSocket.getInetAddress()+" "+pSocket.getLocalAddress());
				pSocket.receive(inboundMesg);
				InetSocketAddress clientSoc = new InetSocketAddress(inboundMesg.getAddress(),inboundMesg.getPort());
				PaxosMessage inboundClientMesg = new PaxosMessage(inboundMesg.getData());
				System.out.println("PID: "+pid+" - Received packet from client: "+inboundClientMesg.senderId);
				clientSocket.set(inboundClientMesg.senderId, clientSoc);
				System.out.println("DEBUG processing clreq");
				protocolState = processClientReq(inboundClientMesg);
			}
			catch (SocketTimeoutException e) 
			{
				// TODO: handle exception
				
			}
			catch (IOException e) 
			{
				// TODO: handle exception
			}

		}while(protocolState == PaxosConsts.CONTINUE);
		
	    return protocolState;
				
	}
	
	public int playClient()
	{
		String clientInp = null;
		ArrayList<String> cmdParamList = new ArrayList<String>();
		do
		{
			System.out.println("DEBUG client pid: "+pid);
			System.out.print("\n Enter command(s): ");
			clientInp = new Scanner(System.in).nextLine();
			System.out.println(clientInp);
			cmdParamList.clear();
			String cmdParam="";
			
			// Split multiple cmd string
			StringTokenizer clientInpTokens = new StringTokenizer(clientInp,";");
			while (clientInpTokens.hasMoreTokens())
			{
				// For each command get command (cmd) and the parameter (param)
				cmdParam = clientInpTokens.nextToken();
				cmdParamList.add(cmdParam);
			}
			
			System.out.println("DEBUG client cmd in list"+cmdParamList.toString());
		
			byte clientOps[] = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CLIENT_REQ, pid, 
					PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_STR, cmdParamList, PaxosConsts.NO_ELE_INT,
					PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,pid); 
			DatagramPacket clientPacket = null; 
			
			try
			{
				clientPacket = new DatagramPacket(clientOps, clientOps.length, InetAddress.getByName(remoteHost),upSet.get(currentLeader));
				System.out.println(" DEBUG "+(new String(clientOps)));
				System.out.println(" DEBUG "+remoteHost);
				System.out.println(" DEBUG "+upSet.get(currentLeader));
			}
			catch (UnknownHostException he) 
			{
				System.err.println(he.getMessage());
			}
			boolean leaderError = false;
			int numRespRec=0;
			do
			{
				System.out.println("DEBUG cmdParamList > 0");
				try
				{
					pSocket.setSoTimeout(PaxosConsts.CLIENT_TIMEOUT_BOUND);
			 		if (!leaderError)
			 		{
			 			System.out.println(" DEBUG size:"+cmdParamList.size());
			 			System.out.println(" DEBUG data:"+cmdParamList.toString());
			 			System.out.println("DEBUG sending client packet");
						System.out.println(" DEBUG "+(new String(clientOps)));
						System.out.println(" DEBUG "+remoteHost);
						System.out.println(" DEBUG "+upSet.get(currentLeader));
			 			pSocket.send(clientPacket);
			 			System.out.println("DEBUG sent client packet to "+clientPacket.getSocketAddress());
			 		}
			 		else
			 		{
			 			// send to all
			 			leaderError = false;
			 			numRespRec = 0;
			 			System.out.println("DEBUG server died so sending to all");
			 			byte resendClientOps[] = (new PaxosMessage()).buildPaxosMsg(PaxosConsts.CLIENT_REQ, pid, 
								PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_INT, PaxosConsts.NO_ELE_STR, cmdParamList, PaxosConsts.NO_ELE_INT,
								PaxosConsts.NO_ELE_STR,PaxosConsts.NO_ELE_STR,pid);
			 			DatagramPacket resendClientPacket = null;
			 			for (int i=0;i<upSet.size();i++)
			 			{
			 				if (upSet.get(i) != PaxosConsts.DEAD_PROCESS_PORT)
			 				{
				 				try
				 				{
				 					resendClientPacket = new DatagramPacket(resendClientOps, resendClientOps.length, InetAddress.getByName(remoteHost),upSet.get(i));
				 					pSocket.send(resendClientPacket);
				 				}
				 				catch (UnknownHostException he) 
				 				{
				 					System.err.println(he.getMessage());
				 				}
			 				}
			 			}
			 		}
					DatagramPacket serverResp = new DatagramPacket(buf,buf.length);
					while(numRespRec < cmdParamList.size() && !leaderError)
					{
						try
						{
							Arrays.fill(serverResp.getData(), 0,PaxosConsts.MESG_SIZE,(byte)0);
							pSocket.receive(serverResp);
							numRespRec++;
							cmdParamList.remove(PaxosConsts.FIRST_REQ);
							PaxosMessage dataFromServer = new PaxosMessage(serverResp.getData());
							System.out.println("DEBUG Got server response: "+dataFromServer.respToClient);
							System.out.println("param list after remove: "+cmdParamList.toString());
							currentLeader = dataFromServer.senderId;
						}
						catch (SocketTimeoutException ste) 
						{
							System.err.println(ste.getMessage());
							leaderError = true;
							upSet.set(currentLeader, PaxosConsts.DEAD_PROCESS_PORT);
							System.out.println("DEBUG server died");
						}
					}
				}
				catch(SocketException se)
				{
					System.err.println(se.getMessage());
					se.printStackTrace();
				}
				catch(IOException io)
				{
					System.err.println(io.getMessage());
					io.printStackTrace();
				}
			}while(cmdParamList.size()>PaxosConsts.NO_ELEMENT);

		}while (clientInp.indexOf(PaxosConsts.EXIT) != -1);
		return PaxosConsts.SESSION_DONE;
	}
	
	public void writeLog(String msg)
	{
		try
		{
			BufferedWriter bw = new BufferedWriter(new FileWriter(logFile));
			bw.write(msg);
			System.out.println("DEBUG"+" wrote to log: "+msg);
		}
		catch(IOException io)
		{
			System.err.println(io.getMessage());
		}
	}
}

